package DataStreamApi.Transformation算子.合流;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;

import java.util.Arrays;
import java.util.List;

/**
 *为了处理更加灵活，连接操作允许流的数据类型不同但我们知道一个DataStream中的数据只能有唯一的类型，所以连接得到的并不是DataStream，
 * 而是一个“连接流连接流可以看成是两条流形式上的“统一”，被放在了一个同一个流中;事实上内部仍保持各自的数据形式不变，彼此之间是相互独立的。
 * 要想得到新的DataStream还需要进一步定义一个“同处理”转换操(co-process)作，用来说明对于不同来源、不同类型的数据，怎样分别进行处理转换、
 * 得到统一的输出类型。所以整体上来，两条流的连接就像是“一国两制”两条流可以保持各自的数据类型、处理方式也可以不同不过最终还是会统一到同一个DataStream中。
 */
public class Flink02_Connect {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment executionEnvironment = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());


        List<Integer> list1 = Arrays.asList(
                1, 2, 3, 4, 5, 6, 7);


        List<String> list3 = Arrays.asList(
                "5", "6", "7");
        executionEnvironment.setParallelism(6);
        DataStreamSource<Integer> i1 = executionEnvironment.fromCollection(list1);
        DataStreamSource<String> s1 = executionEnvironment.fromCollection(list3);

        ConnectedStreams<Integer, String> connect = i1.connect(s1);

        SingleOutputStreamOperator<Integer> map = connect.map(new CoMapFunction<Integer, String, Integer>() {
            @Override
            public Integer map1(Integer value) throws Exception {
                return value;
            }

            @Override
            public Integer map2(String value) throws Exception {
                return Integer.parseInt(value);
            }
        });


        map.print();
        executionEnvironment.execute();
    }
}
